// Copyright (c) 2020-present, INSPUR Co, Ltd. All rights reserved.
// This source code is licensed under Apache 2.0 License.

#include <limits>
#include "pure_mem/rowex_art.h"

namespace rocksdb {
// convert Slice key object to Key object used in ART tree.
void ParseSlice2Key(const Slice& userKey, art_rowex::Key& key) {
  key.set(userKey.data(), userKey.size());
}

// parse TID with ART tree leaf node , generate Key object.
void ParseTid2Key(TID tid, art_rowex::Key& key) {
  auto ref = reinterpret_cast<ARTRef*>(tid);
  while (ref->getContentList() == nullptr) {
    ref = ref->Next();
    if (ref == nullptr) return;
  }
  auto storeNode = static_cast<VersionNode*>(ref->getContentList());
  MvccKey mk;
  mk.parseKey(GetLengthPrefixedSlice(storeNode->Key()));
  key.set(mk.userKey_.data(), mk.userKey_.size());
}

// use c-type type convert function.
inline ARTRef* ForceTID2Ref(TID& t) { return reinterpret_cast<ARTRef*>(t); }

ARTIndex4Mem::ARTIndex4Mem() {
  tree_ = new art_rowex::Tree(ParseTid2Key);
  artref_dlink_ = new ConcurrentDLink<ARTRef*>();
}

ARTIndex4Mem::~ARTIndex4Mem() {
  ARTRef* cur = artref_dlink_->getHead(), *next;
  while(cur != nullptr){
    next = cur->Next();
    delete cur;
    cur = next;
  }

  delete tree_;
}


ARTRef* ARTIndex4Mem::getOrNewARTRef(Slice& userKey, VersionNode* node) {
  art_rowex::Key key;
  ParseSlice2Key(userKey, key);

  auto curRef = new ARTRef();
  curRef->setContentList(node);
  TID curTID, nextTID;
  // nextTID is not absolutely greater than key, so add compare logic.
  // if ARTRef not exists, we will put it onto ART tree.
  this->tree_->insertOrGetLG(key, (TID)curRef, curTID, nextTID);
  ARTRef* retRef = ForceTID2Ref(curTID);

  // key has ARTRef object on ART tree.
  if (retRef != curRef) {
    delete curRef;
    return retRef;
  }

  ARTRef *nextRef = nullptr;
  if (nextTID != 0) {
    nextRef = ForceTID2Ref(nextTID);
    if ((nextRef->Next() == nullptr || nextRef->Prev() == nullptr) &&
        this->tree_->compareKeyPrefix(nextTID, key,
                                      std::numeric_limits<int>::max()) <= 0) {
      // next < key  and  next has not inserted into double link list.
      // need redo get larger ARTRef object from ART tree.
      nextTID = this->tree_->getLG(key);
      if (nextTID != 0)
        nextRef = ForceTID2Ref(nextTID);
      else {
        nextRef = nullptr;
      }
    }
  }
  if (nextRef == nullptr) nextRef = artref_dlink_->getTail();
  // insert current ARTRef object into double link list.
  art_rowex::Tree* treecomp = this->tree_;
  artref_dlink_->insert(retRef, nextRef, [&key, treecomp](ARTRef* a) -> int {
    return treecomp->compareKeyPrefix((TID)a, key,
                                      std::numeric_limits<int>::max());
  });

  return retRef;
}

ARTRef* ARTIndex4Mem::getEGARTRef(Slice& userKey) const{
  art_rowex::Key key;
  ParseSlice2Key(userKey, key);
  TID eorg = this->tree_->getEG(key);
  ARTRef* retRef = nullptr;
  if (eorg != 0) {
    retRef = ForceTID2Ref(eorg);
  }

  return retRef;
}

}